package com.qingcloud.df.executor.config;

import com.qingcloud.df.executor.worker.ExecutorHeartbeatJob;
import com.qingcloud.df.executor.worker.ExecutorService;
import com.qingcloud.df.executor.worker.NodeService;
import com.qingcloud.df.sdk.component.config.ExecutorConfig;
import com.qingcloud.df.sdk.component.context.ConfigManager;
import com.qingcloud.df.sdk.component.init.NodeRegister;
import com.qingcloud.df.sdk.component.init.ServerPing;
import com.qingcloud.df.sdk.component.logger.CompInstFileAppender;
import com.qingcloud.df.sdk.component.utils.ComponentCommUtils;
import com.qingcloud.df.sdk.component.utils.ThreadExecutorManager;
import org.quartz.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.context.annotation.Configuration;
import javax.annotation.Resource;
import java.util.concurrent.TimeUnit;

@Configuration
public class ExecutorRunner implements CommandLineRunner {
    public static final Logger logger = LoggerFactory.getLogger(ExecutorRunner.class);

    @Value("${server.port}")
    private int serverPort;
    @Value("${qingcloud.executor.env}")
    private String env;
    @Value("${qingcloud.executor.type}")
    private String type;
    @Value("${qingcloud.executor.server}")
    private String serverAddress;
    @Value("${qingcloud.executor.bucketName}")
    private String bucketName;
    @Value("${qingcloud.executor.workspaceRootPath}")
    private String workspaceRootPath;
    @Value("${qingcloud.executor.componentDownload}")
    private boolean componentDownload;
    @Value("${qingcloud.executor.task.type}")
    private String taskType;
    @Value("${qingcloud.executor.akka.node}")
    private String akkaNodeAddress;
    @Value("${qingcloud.executor.akka.server}")
    private String akkaServerAddress;
    @Value("${qingcloud.executor.instance.runtime}")
    private String runtime;
    @Value("${qingcloud.executor.instance.configDownload}")
    private boolean configDownload;
    @Value("${qingcloud.executor.instance.maxLightJobNum}")
    private Integer maxLightJobNum;
    @Value("${qingcloud.executor.instance.maxHeavyJobNum}")
    private Integer maxHeavyJobNum;
    @Value("${qingcloud.executor.heartbeatCron}")
    private String heartbeatCron ="0/30 * * ? * * *";      //网关心跳数据上报时间
    @Resource
    AkkaConfig akkaConfig;
    @Resource
    private NodeService nodeService;
    @Resource
    private Scheduler scheduler;
    @Override
    public void run(String... args) {
        //设置环境参数
        CompInstFileAppender.initLogPath(ComponentCommUtils.getLogsBaseDir());
        //启动所有组件实例
        try {
            ExecutorConfig.setType(type); //节点类型  server/executor/gateway
            ExecutorConfig.setServerAddress(serverAddress);//集群Server address
            ExecutorConfig.setAkkaNodeAddress(akkaNodeAddress);    //当前节点node 地址端口
            ExecutorConfig.setAkkaServerAddress(akkaServerAddress);
            ExecutorConfig.setEnv(env);
            ExecutorConfig.setServerPort(serverPort);
            ExecutorConfig.setConfigDownload(configDownload);
            ExecutorConfig.setBucketName(bucketName);
            ExecutorConfig.setComponentDownload(componentDownload);
            ExecutorConfig.setRuntime(runtime);
            ExecutorConfig.setTaskType(taskType);
            ExecutorConfig.setWorkspaceRootPath(workspaceRootPath);
            ExecutorConfig.setMaxHeavyJobNum(maxHeavyJobNum);
            ExecutorConfig.setMaxLightJobNum(maxLightJobNum);
            logger.info("executor system init complete");

            String[] nodeTypes =  type.split("#");
            //初始化AKka
            akkaConfig.init();
            ConfigManager.init();
            for(String nodeType : nodeTypes) {
                if (nodeType.equals("server")) {
                    logger.info("Server startup successful!");
                    //初始化AKka
//                    try {
//                        ClusterManager.init();
//                    } catch (Exception ex) {
//                        logger.error("executor initialize failed!", ex);
//                    }
                } else if (nodeType.equals("executor")) {
                    ThreadExecutorManager executorManager = new ThreadExecutorManager();
                    executorManager.getRegisterExecutor().scheduleWithFixedDelay(() -> {
                        //确保先注册成功，否则一一直循环知道注册成功后再启动相关服务
                        boolean register = doExecutorRegister();
                        if (register) {
                            try {
                                executorManager.getRegisterExecutor().shutdown();
                                //执行节点初始化
                                nodeService.init();
                                logger.info("Executor startup successful!");
                            } catch (Exception e) {
                                throw new RuntimeException(e);
                            }
                        }
                    }, 0, 5, TimeUnit.SECONDS);

                    /**
                     * 执行器心跳
                     */
                    createJob("ExecutorNodeGroup", "executorHeartbeatJob", heartbeatCron, ExecutorHeartbeatJob.class);
                }
            }
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
    }

    private synchronized boolean doExecutorRegister()  {
        try {
            //确认是否连通到Server
            ServerPing.init();
            //节点注册到Server
            NodeRegister.init();
        }catch (Throwable ex) {
            logger.error("executor initialize failed!", ex);
            return false;
        }
        return true;
    }


    /**
     * 创建调度任务
     *
     * @param group    group
     * @param name     name
     * @param corn     corn
     * @param jobClass class
     */
    private void createJob(String group, String name, String corn, Class<? extends Job> jobClass) throws SchedulerException {
        JobDetail jobDetail = JobBuilder.newJob(jobClass).withIdentity(name, group).build();
        Trigger trigger = TriggerBuilder.newTrigger()
                .withIdentity(name, group)
                .startAt(DateBuilder.futureDate(1, DateBuilder.IntervalUnit.SECOND))
                .withSchedule(CronScheduleBuilder.cronSchedule(corn))
                .startNow().build();
        scheduler.scheduleJob(jobDetail, trigger);
    }

}
